package com.haier.spark.project.spark

import java.util.Date

import com.haier.spark.project.conf.ConfigurationManager
import com.haier.spark.project.constant.Constants
import com.haier.spark.project.dao.impl.{SessionAggrStatDaoImpl, SessionDetailDaoImpl, SessionRandomExtractDaoImpl, TaskDaoImpl}
import com.haier.spark.project.domain.{SessionAggrStat, SessionDetail, SessionRandomExtract}
import com.haier.spark.project.util._
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}
import org.slf4j.LoggerFactory

import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.util.Random


/**
  * 用户访问session分析
  */
object UserVisitSessionAnalyze {
  private val logger = LoggerFactory.getLogger(getClass)

  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = getSparkSession
    if (ConfigurationManager.getBoolean(Constants.SPARK_LOCAL)) {
      TestMockData.mock(sparkSession)
    }

    //    val taskId: Long = ParamUtils.getTaskIdFromArgs(args)
    val taskId: Long = 1
    val task = new TaskDaoImpl().findById(taskId)
    val actionRDD = getActionRDDByDateRange(sparkSession, task.taskParam)
    println("actionRDD count = " + actionRDD.count())
    actionRDD.take(10).foreach(println)

    val sessionId2ActionRDD: RDD[(String, Row)] = actionRDD.map(row => {
      (row.getString(2), row)
    })
    // (sessionId, key1=xxx|key2=xxx)
    val sessionId2AggrInfoRDD = aggregateBySession(sparkSession, sessionId2ActionRDD)
    println("sessionId2AggrInfoRDD count = " + sessionId2AggrInfoRDD.count())
    sessionId2AggrInfoRDD.take(10).foreach(println)

    val sessionAggrStatAccumulator = new SessionAggrStatAccumulator()

    /**
      * 使用Accumulator必须注册，否则会报Task not Serializable的错误
      */
    sparkSession.sparkContext.register(sessionAggrStatAccumulator)
    // (sessionId, key1=xxx|key2=xxx)
    val filterSessionRDD = filterSessionAndStatistic(sessionId2AggrInfoRDD, task.taskParam, sessionAggrStatAccumulator)
    println("filterSessionRDD count = " + filterSessionRDD.count())
    filterSessionRDD.take(10).foreach(println)

    /**
      * Accumulator.value使用之前必须有action操作，否则获取不到值
      */
    calculateAndPersistAggrStat(sessionAggrStatAccumulator.value, task.taskId)

    randomExtractSession(filterSessionRDD, taskId, sessionId2ActionRDD)

    val sessionId2DetailRDD: RDD[(String, Row)] = getSessionId2DetailRDD(filterSessionRDD, sessionId2ActionRDD)
    val top10CategoryMap: Map[Long, (Long, Long, Long)] = getTop10Category(taskId, sessionId2DetailRDD)

    getTop10Session(taskId, sessionId2DetailRDD, top10CategoryMap, sparkSession.sparkContext)
    /*
     session聚合统计（统计出访问时长和访问步长，各个区间的session数量占总session数量的比例）
     开发步骤（未进行优化）
     1.将actionRDD映射为RDD[sessionId, Row]格式
     2.根据sessionId聚合，计算该sessionId对应的访问时间和访问步长，生成一个新的RDD
     3.遍历新生成的RDD，将访问时间和访问步长更新到自定义的Accumulator中
     4.使用自定义的Accumulator去统计，计算出各个区间的比例
     5.将统计结果插入数据库中
    val testRDD = actionRDD.map(row => (row.getString(2), row)).groupByKey().mapValues(f = rows => {
      var time: Int = 0
      var startTime: String = null
      var endTime: String = null
      val step = rows.size
      rows.foreach(row => {
        val actionTime = row.getString(4)
        if (startTime == null || startTime.compareTo(actionTime) > 0) {
          startTime = actionTime
        }
        if (endTime == null || endTime.compareTo(actionTime) < 0) {
          endTime = actionTime
        }
      })
      time = DateUtils.minus(startTime, endTime)
      (time, step)
    })
    优化：
    重构sessionId2AggrInfoRDD，遍历actionRDD时，顺便计算出访问时间和访问步长，拼接到value中
      */

    sparkSession.close()
  }

  /**
    * 获取指定日期范围内的用户访问行为数据
    *
    * @param sparkSession sparkSession
    * @param taskParam    任务参数
    * @return
    */
  private def getActionRDDByDateRange(sparkSession: SparkSession, taskParam: String): RDD[Row] = {
    val startDate = ParamUtils.getParam(taskParam, "startDate")
    val endDate = ParamUtils.getParam(taskParam, "endDate")
    val sql = s"select * from user_visit_action where date >= '$startDate' and date <= '$endDate'"
    val actionDataFrame = sparkSession.sql(sql)
    actionDataFrame.rdd
  }

  /**
    * 对行为数据根据session粒度聚合
    *
    * @param sparkSession        sparkSession
    * @param sessionId2ActionRDD sessionId2ActionRDD
    * @return
    */
  private def aggregateBySession(sparkSession: SparkSession, sessionId2ActionRDD: RDD[(String, Row)]): RDD[(String, String)] = {
    // 将同一个sessionId的行为数据聚合到一起
    val sessionId2ActionsRDD = sessionId2ActionRDD.groupByKey()

    //搜索某些关键词的用户、访问时间在某个时间段内的用户、年龄在某个范围内的用户、职业在某个范围内的用户、所在某个城市的用户，发起的session。
    val userId2PartAggrInfoRDD = sessionId2ActionsRDD.map(item => {
      val rows = item._2
      val userId = rows.head.getLong(1)
      val sessionId = item._1
      val searchKeywords: StringBuilder = new StringBuilder()
      val clickCategoryIds: StringBuilder = new StringBuilder()
      var startTime: Date = null
      var endTime: Date = null
      rows.foreach(f = row => {
        // 搜索词跳转
        val searchKeyword = row.getString(5)
        var clickCategoryId: Long = -1
        // 点击产品跳转
        try {
          clickCategoryId = row.getLong(6)
        } catch {
          case _: NullPointerException =>
            clickCategoryId = -1
        }

        // 搜索词和页面点击只可能同时存在一个
        if (searchKeyword != null && !searchKeyword.eq("")) {
          if (!searchKeywords.toString().contains(searchKeyword)) {
            searchKeywords.append(searchKeyword).append(",")
          }
        } else if (clickCategoryId >= 0) {
          if (!clickCategoryIds.toString().contains(clickCategoryId.toString)) {
            clickCategoryIds.append(clickCategoryId.toString).append(",")
          }
        }

        // 统计session的访问时间
        val actionTime = DateUtils.parseTime(row.getString(4))
        if (null == startTime || startTime.after(actionTime)) {
          startTime = actionTime
        }
        if (null == endTime || endTime.before(actionTime)) {
          endTime = actionTime
        }
      })
      val step = rows.size
      val visitTime = (endTime.getTime - startTime.getTime) / 1000

      val partAggrInfo = new StringBuilder(Constants.FIELD_SESSION_ID).append("=").append(sessionId).append("|")
        .append(Constants.FIELD_SEARCH_KEYWORDS).append("=").append(searchKeywords).append("|")
        .append(Constants.FIELD_CLICK_CATEGORY_IDS).append("=").append(clickCategoryIds).append("|")
        .append(Constants.FIELD_VISIT_TIME).append("=").append(visitTime).append("|")
        .append(Constants.FIELD_STEP_LENGTH).append("=").append(step).append("|")
        .append(Constants.FIELD_START_TIME).append("=").append(DateUtils.formatTime(startTime))
      (userId, partAggrInfo.toString())
    })

    val userInfoRDD = getUserInfoRDD(sparkSession)

    val sessionId2FullAggrInfoRDD = userId2PartAggrInfoRDD.join(userInfoRDD).map(item => {
      val partAggrInfo = item._2._1
      val userRow = item._2._2
      val sessionId = StringUtils.getFieldFromConcatString(partAggrInfo, "\\|", Constants.FIELD_SESSION_ID)
      val age = userRow.getInt(3)
      val professional = userRow.getString(4)
      val city = userRow.getString(5)
      val sex = userRow.getString(6)
      val fullAggrInfo = new StringBuilder(partAggrInfo).append("|")
        .append(Constants.FIELD_AGE).append("=").append(age).append("|")
        .append(Constants.FIELD_PROFESSIONAL).append("=").append(professional).append("|")
        .append(Constants.FIELD_CITY).append("=").append(city).append("|")
        .append(Constants.FIELD_SEX).append("=").append(sex)
      (sessionId, fullAggrInfo.toString)
    })
    sessionId2FullAggrInfoRDD
  }

  /**
    * 获取用户信息的RDD
    *
    * @param sparkSession sparkSession
    * @return
    */
  private def getUserInfoRDD(sparkSession: SparkSession): RDD[(Long, Row)] = {
    val sql = "select * from user_info"
    val userDF = sparkSession.sql(sql)
    userDF.rdd.map(row => {
      (row.getLong(0), row)
    })
  }

  /**
    * 根据条件过滤并且统计访问时间和访问步长
    *
    * @param sessionId2AggrInfoRDD      sessionId和session信息对应的RDD
    * @param taskParam                  任务参数
    * @param sessionAggrStatAccumulator 自定义Accumulator
    * @return
    */
  private def filterSessionAndStatistic(sessionId2AggrInfoRDD: RDD[(String, String)], taskParam: String,
                                        sessionAggrStatAccumulator: SessionAggrStatAccumulator): RDD[(String, String)] = {
    // 将查询参数拼接成需要的格式，方便后续过滤使用
    val formatParam = getFormatTaskParam(taskParam)
    val filteredSessionid2AggrInfoRDD = sessionId2AggrInfoRDD.filter(f = item => {
      val aggrInfo = item._2
      // 按照年龄范围进行过滤（startAge、endAge）
      val result = ValidUtils.between(aggrInfo, Constants.FIELD_AGE, formatParam, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE) &&
        // 按照职业范围进行过滤（professionals）
        // 互联网,IT,软件
        // 互联网
        ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL, formatParam, Constants.PARAM_PROFESSIONALS) &&
        // 按照城市范围进行过滤（cities）
        // 北京,上海,广州,深圳
        // 成都
        ValidUtils.in(aggrInfo, Constants.FIELD_CITY, formatParam, Constants.PARAM_CITIES) &&
        // 按照性别进行过滤
        // 男/女
        // 男，女
        ValidUtils.equal(aggrInfo, Constants.FIELD_SEX, formatParam, Constants.PARAM_SEX) &&
        // 按照搜索词进行过滤
        // 我们的session可能搜索了 火锅,蛋糕,烧烤
        // 我们的筛选条件可能是 火锅,串串香,iphone手机
        // 那么，in这个校验方法，主要判定session搜索的词中，有任何一个，与筛选条件中
        // 任何一个搜索词相当，即通过
        ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS, formatParam, Constants.PARAM_KEYWORDS) &&
        // 按照点击品类id进行过滤
        ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS, formatParam, Constants.PARAM_CATEGORY_IDS)
      if (result) {
        // session 符合任务的过滤条件，统计时间和步长
        val stepLength = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH).toLong
        val visitTime = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_VISIT_TIME).toLong
        sessionAggrStatAccumulator.add(Constants.SESSION_COUNT)
        calcStepLent(stepLength)
        calcVisitTime(visitTime)
      }

      /**
        * 计算访问时间范围
        *
        * @param visitTime 访问时长
        */
      def calcVisitTime(visitTime: Long): Unit = {
        if (visitTime >= 1 && visitTime <= 3) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s)
        } else if (visitTime >= 4 && visitTime <= 6) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s)
        } else if (visitTime >= 7 && visitTime <= 9) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s)
        } else if (visitTime >= 10 && visitTime <= 30) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s)
        } else if (visitTime > 30 && visitTime <= 60) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s)
        } else if (visitTime > 60 && visitTime <= 180) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m)
        } else if (visitTime > 180 && visitTime <= 600) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m)
        } else if (visitTime > 600 && visitTime <= 1800) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m)
        } else if (visitTime > 1800) {
          sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m)
        }
      }

      /**
        * 计算访问步长范围
        *
        * @param stepLength 访问步长
        */
      def calcStepLent(stepLength: Long): Unit = {
        if (stepLength >= 1 && stepLength <= 3) {
          sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3)
        } else if (stepLength >= 4 && stepLength <= 6) {
          sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6)
        } else if (stepLength >= 7 && stepLength <= 9) {
          sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9)
        } else if (stepLength >= 10 && stepLength <= 30) {
          sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30)
        } else if (stepLength > 30 && stepLength <= 60) {
          sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60)
        } else if (stepLength > 60) {
          sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60)
        }
      }

      result
    })

    filteredSessionid2AggrInfoRDD
  }

  private def getFormatTaskParam(taskParam: String): String = {
    val startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE)
    val endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE)
    val professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS)
    val cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES)
    val sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX)
    val keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS)
    val categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS)
    var formatParam = (if (startAge != null) Constants.PARAM_START_AGE + "=" + startAge + "|"
    else "") + (if (endAge != null) Constants.PARAM_END_AGE + "=" + endAge + "|"
    else "") + (if (professionals != null) Constants.PARAM_PROFESSIONALS + "=" + professionals + "|"
    else "") + (if (cities != null) Constants.PARAM_CITIES + "=" + cities + "|"
    else "") + (if (sex != null) Constants.PARAM_SEX + "=" + sex + "|"
    else "") + (if (keywords != null) Constants.PARAM_KEYWORDS + "=" + keywords + "|"
    else "") + (if (categoryIds != null) Constants.PARAM_CATEGORY_IDS + "=" + categoryIds
    else "")
    if (formatParam.endsWith("\\|")) {
      formatParam = formatParam.substring(0, formatParam.length - 1)
    }
    formatParam
  }

  /**
    * 计算各session范围占比，并写入MySQL
    *
    * @param value  自定义Accumulator value
    * @param taskId 任务ID
    */
  private def calculateAndPersistAggrStat(value: String, taskId: Long) = {
    // 从Accumulator统计串中获取值
    val session_count = StringUtils.getFieldFromConcatString(value, "\\|", Constants.SESSION_COUNT).toDouble

    val visit_length_1s_3s = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_1s_3s).toLong
    val visit_length_4s_6s = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_4s_6s).toLong
    val visit_length_7s_9s = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_7s_9s).toLong
    val visit_length_10s_30s = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_10s_30s).toLong
    val visit_length_30s_60s = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_30s_60s).toLong
    val visit_length_1m_3m = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_1m_3m).toLong
    val visit_length_3m_10m = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_3m_10m).toLong
    val visit_length_10m_30m = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_10m_30m).toLong
    val visit_length_30m = StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_30m).toLong

    val step_length_1_3 = StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_1_3).toLong
    val step_length_4_6 = StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_4_6).toLong
    val step_length_7_9 = StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_7_9).toLong
    val step_length_10_30 = StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_10_30).toLong
    val step_length_30_60 = StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_30_60).toLong
    val step_length_60 = StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_60).toLong

    // 计算各个访问时长和访问步长的范围
    val visit_length_1s_3s_ratio = NumberUtils.formatDouble(visit_length_1s_3s.toDouble / session_count, 2)
    val visit_length_4s_6s_ratio = NumberUtils.formatDouble(visit_length_4s_6s.toDouble / session_count, 2)
    val visit_length_7s_9s_ratio = NumberUtils.formatDouble(visit_length_7s_9s.toDouble / session_count, 2)
    val visit_length_10s_30s_ratio = NumberUtils.formatDouble(visit_length_10s_30s.toDouble / session_count, 2)
    val visit_length_30s_60s_ratio = NumberUtils.formatDouble(visit_length_30s_60s.toDouble / session_count, 2)
    val visit_length_1m_3m_ratio = NumberUtils.formatDouble(visit_length_1m_3m.toDouble / session_count, 2)
    val visit_length_3m_10m_ratio = NumberUtils.formatDouble(visit_length_3m_10m.toDouble / session_count, 2)
    val visit_length_10m_30m_ratio = NumberUtils.formatDouble(visit_length_10m_30m.toDouble / session_count, 2)
    val visit_length_30m_ratio = NumberUtils.formatDouble(visit_length_30m.toDouble / session_count, 2)

    val step_length_1_3_ratio = NumberUtils.formatDouble(step_length_1_3.toDouble / session_count, 2)
    val step_length_4_6_ratio = NumberUtils.formatDouble(step_length_4_6.toDouble / session_count, 2)
    val step_length_7_9_ratio = NumberUtils.formatDouble(step_length_7_9.toDouble / session_count, 2)
    val step_length_10_30_ratio = NumberUtils.formatDouble(step_length_10_30.toDouble / session_count, 2)
    val step_length_30_60_ratio = NumberUtils.formatDouble(step_length_30_60.toDouble / session_count, 2)
    val step_length_60_ratio = NumberUtils.formatDouble(step_length_60.toDouble / session_count, 2)

    val sessionAggrStat = SessionAggrStat(taskId, session_count,
      visit_length_1s_3s_ratio, visit_length_4s_6s_ratio, visit_length_7s_9s_ratio,
      visit_length_10s_30s_ratio, visit_length_30s_60s_ratio,
      visit_length_1m_3m_ratio, visit_length_3m_10m_ratio,
      visit_length_10m_30m_ratio, visit_length_30m_ratio,
      step_length_1_3_ratio, step_length_4_6_ratio,
      step_length_7_9_ratio, step_length_10_30_ratio,
      step_length_30_60_ratio, step_length_60_ratio)

    val sessionDao = new SessionAggrStatDaoImpl()
    sessionDao.insert(sessionAggrStat)
  }

  /**
    * 随机抽取session
    *
    * @param filterSessionRDD 经过过滤后的sessionRDD
    * @return
    */
  private def randomExtractSession(filterSessionRDD: RDD[(String, String)], taskId: Long, sessionId2ActionRDD: RDD[(String, Row)]): Unit = {
    //(yyyy-MM-dd_HH,key1=xxx|key2=xxx)
    val time2SessionRDD: RDD[(String, String)] = filterSessionRDD.map(item => {
      val aggrInfo = item._2
      val startTime = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_START_TIME)
      val dayAndHour = DateUtils.getDateHour(startTime)
      (dayAndHour, aggrInfo)
    })
    // 1.获取每天每小时的session数量 (yyyy-MM-dd_HH -> 100)
    val dayCountMap: collection.Map[String, Long] = time2SessionRDD.countByKey()

    // 2.随机抽取算法
    val dayHourExtractMap: collection.Map[String, collection.Map[String, ListBuffer[Int]]] = randomExtract(dayCountMap, 100)

    // 3.随机抽取出session，并保存到数据库中，同时生成新的RDD（包含抽取出来的sessionId）
    val time2SessionsRDD: RDD[(String, Iterable[String])] = time2SessionRDD.groupByKey()
    val extractSessionIdsRDD = time2SessionsRDD.flatMap(item => {
      val day = item._1.split("_")(0)
      val hour = item._1.split("_")(1)
      val aggrInfoList: List[String] = item._2.toList
      val sessionRandomExtractDao = new SessionRandomExtractDaoImpl()
      dayHourExtractMap(day)(hour).map(index => {
        val aggrInfo: String = aggrInfoList(index)
        println("aggreInfo = " + aggrInfo)
        val sessionId = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_SESSION_ID)
        // 保存到数据库
        val sessionRandomExtract: SessionRandomExtract = SessionRandomExtract(taskId, sessionId,
          StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_START_TIME),
          StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS),
          StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS))
        sessionRandomExtractDao.insert(sessionRandomExtract)
        (sessionId, sessionId)
      })
    })

    // 4.根据sessionId关联session详情信息，将sessionDetail保存到数据库中
    extractSessionIdsRDD.join(sessionId2ActionRDD).foreachPartition(partition => {
      //TODO 需要修改，如果该步骤放到foreachPartition外边，会报Task not Serializable的错误，原因暂未找到
      //      val sessionDetailDao = new SessionDetailDaoImpl()
      partition.foreach(record => {
        val row: Row = record._2._2
        val sessionDetail = SessionDetail(
          taskId, row.getLong(1), row.getString(2), row.getLong(3), row.getString(4), row.getString(5), row.getLong(6), row.getLong(7),
          row.getString(8), row.getString(9), row.getString(10), row.getString(11)
        )

        // TODO 此处的数据库连接会超出上限，暂未处理
        //        sessionDetailDao.insert(sessionDetail)
      })
    })
  }

  /**
    * 获取top10热门品类并保存到数据库中
    *
    * @param taskId              任务id
    * @param sessionId2DetailRDD 符合条件的sessionId和对应的session信息RDD
    * @return top10CategoryMap (categoryId -> (clickCount, orderCount, payCount))
    */
  private def getTop10Category(taskId: Long, sessionId2DetailRDD: RDD[(String, Row)]): Map[Long, (Long, Long, Long)] = {
    // 1.获取所有符合条件的session访问过的所有品类

    //2. 获取各个品类数量
    // 点击数量
    val clickCategoryId2CountRDD: RDD[(Long, Long)] = sessionId2DetailRDD.filter(_._2.getLong(6) >= 0).map(_._2.getLong(6) -> 1L).reduceByKey(_ + _)
    // 下单数量
    val orderedCategoryId2CountRdd: RDD[(Long, Long)] = sessionId2DetailRDD
      .filter(item => {
        item._2.getString(8) != null && !item._2.getString(8).eq("")
      })
      .flatMap(item => {
        item._2.getString(8).split(",").map(_.toLong -> 1L)
      }).reduceByKey(_ + _)
    // 支付数量
    val payCategoryId2CountRdd: RDD[(Long, Long)] = sessionId2DetailRDD
      .filter(item => {
        item._2.getString(10) != null && !item._2.getString(10).eq("")
      })
      .flatMap(item => {
        item._2.getString(10).map(_.toLong -> 1L)
      }).reduceByKey(_ + _)

    // 所有的品类ID,必须进行去重操作
    val categoryIdRDD: RDD[(Long, Long)] = sessionId2DetailRDD.flatMap(item => {
      val row: Row = item._2
      val categoryId: Long = row.getLong(6)
      val array: ArrayBuffer[(Long, Long)] = new ArrayBuffer[(Long, Long)]()
      if (null != categoryId) {
        array.append((categoryId, categoryId))
      }
      val orderCategoryIds = row.getString(8)
      if (null != orderCategoryIds && !orderCategoryIds.eq("")) {
        orderCategoryIds.split(",").foreach(id => {
          if (null != id) {
            array.append((id.toLong, id.toLong))
          }
        })
      }

      val payCategoryIds = row.getString(10)
      if (null != payCategoryIds && !payCategoryIds.eq("")) {
        payCategoryIds.split(",").foreach(id => {
          if (null != id) {
            array.append((id.toLong, id.toLong))
          }
        })
      }
      array
    }).distinct()

    //3.join各品类与它的点击、下单和支付的次数(categoryId, (clickCount,orderCount,payCount))
    val categoryId2CountRDD: RDD[(Long, (Long, Long, Long))] = categoryIdRDD
      // (categoryId, (categoryId, Option(clickCount)))
      .leftOuterJoin(clickCategoryId2CountRDD)
      // (categoryId, clickCount)
      .map(item => (item._1, item._2._2.getOrElse(0L)))
      // (categoryId, (clickCount, Option(orderCount)))
      .leftOuterJoin(orderedCategoryId2CountRdd)
      // (categoryId, (clickCount, orderCount))
      .map(item => {
      (item._1, (item._2._1, item._2._2.getOrElse(0L)))
    })
      // (categoryId, ((clickCount,orderCount), Option[Long]))
      .leftOuterJoin(payCategoryId2CountRdd)
      // (categoryId, (clickCount,orderCount,payCount))
      .map(item => (item._1, (item._2._1._1, item._2._1._2, item._2._2.getOrElse(0L))))

    //4.排序
    val sortedRDD: RDD[(Long, (Long, Long, Long))] = categoryId2CountRDD.sortBy(item => (item._2._1, item._2._2, item._2._3), ascending = false)
    sortedRDD.take(10).foreach(println)
    sortedRDD.take(10).toMap

    // 自定义排序规则，个人感觉没有sortBy好用
    //    categoryId2CountRDD.map(item => (new CategorySort(item._2._1, item._2._2, item._2._3),item._1)).sortByKey(ascending = false)
    //      .foreach(item => println((item._2, (item._1.clickCount, item._1.orderCount, item._1.payCount))))
  }


  /**
    * 获取top10活跃session
    *
    * @param taskId 任务id
    * @param sessionId2DetailRDD
    */
  def getTop10Session(taskId: Long, sessionId2DetailRDD: RDD[(String, Row)], top10CategoryMap: Map[Long, (Long, Long, Long)], sparkContext: SparkContext): Unit = {
    // 1.获取top10热门品类的id，生成RDD (categoryId, categoryId)
    val top10CategoryIdRDD: RDD[(Long, Long)] = sparkContext.parallelize(top10CategoryMap.keys.map(item => (item, item)).toList)
    // 2.计算top10热门品类被各个session点击的次数 (categoryId, (sessionCount, sessionId))
    val categoryId2SessionCountRDD: RDD[(Long, (Long, String))] = sessionId2DetailRDD.groupByKey()
      .flatMap(item => {
        val sessionId = item._1
        val categoryCountMap: mutable.Map[Long, Long] = mutable.Map()
        item._2.foreach(row => {
          val categoryId = row.getLong(6)
          var count: Long = categoryCountMap.getOrElse(categoryId, 0L)
          count += 1L
          categoryCountMap += (categoryId -> count)
        })
        categoryCountMap.map(row => {
          (row._1, (row._2, sessionId))
        })
      })

    val top10CategoryId2SessionCountRDD: RDD[(Long, (Long,String))] = top10CategoryIdRDD.join(categoryId2SessionCountRDD).map(item => (item._1, item._2._2))
    top10CategoryId2SessionCountRDD.foreach(println)
    // 3.分组取topN算法实现，获取每个热门品类的活跃session
    // 4.获取top10活跃session的明细数据，写入数据库
  }

  /**
    * 获取所有符合条件的sessionId和对应的session信息RDD
    *
    * @param filterSessionRDD    过滤过后的session
    * @param sessionId2ActionRDD sessionId和session信息RDD
    * @return
    */
  private def getSessionId2DetailRDD(filterSessionRDD: RDD[(String, String)], sessionId2ActionRDD: RDD[(String, Row)]) = {
    filterSessionRDD
      // (sessionId, (sessionId, Row))
      .join(sessionId2ActionRDD)
      // (sessionId, Row)
      .map(e => (e._1, e._2._2))
  }

  /**
    * 随机抽取算法
    *
    * @param dayCountMap  (yyyy-MM-dd_HH -> 100)
    * @param extractCount 抽取数量
    */
  private def randomExtract(dayCountMap: collection.Map[String, Long], extractCount: Int): collection.Map[String, collection.Map[String, ListBuffer[Int]]] = {
    // all session counts
    val sumCounts: Long = dayCountMap.values.sum
    /**
      * 存放一天中各个小时的session数量 (yyyy-MM-dd -> (HH -> count))
      */
    var dayHourCountMap: mutable.Map[String, mutable.Map[String, Long]] = mutable.Map()

    // 遍历所有的数据
    dayCountMap.foreach(item => {
      val day: String = item._1.split("_")(0)
      val hour: String = item._1.split("_")(1)
      val count: Long = item._2

      var hourCountMap: mutable.Map[String, Long] = dayHourCountMap.getOrElse(day, null)
      if (null == hourCountMap) {
        hourCountMap = mutable.Map()
        dayHourCountMap += (day -> hourCountMap)
      }
      hourCountMap += (hour -> count)
    })

    println("dayHourCountMap" + dayHourCountMap)
    /**
      * 存放每天每小时需要抽取的session的下标 (yyyy-MM-dd -> (HH -> (5,8,10)))
      */
    var dayHourExtarctMap: mutable.Map[String, mutable.Map[String, ListBuffer[Int]]] = mutable.Map()

    val random: Random = new Random()
    // 遍历每天的数据
    dayHourCountMap.foreach(dayHourCount => {
      val day: String = dayHourCount._1
      val hourCountMap: mutable.Map[String, Long] = dayHourCount._2
      // every day session counts
      val daySessionCounts: Long = hourCountMap.values.sum
      var hourExtractMap: mutable.Map[String, ListBuffer[Int]] = dayHourExtarctMap.getOrElse(day, null)
      if (null == hourExtractMap) {
        hourExtractMap = mutable.Map()
        dayHourExtarctMap += (day -> hourExtractMap)
      }
      // 每天需要抽取的数量
      val dayExtractCount: Int = (daySessionCounts.toDouble / sumCounts * extractCount).toInt
      //      val dayExtractCount: Int = extractCount / dayHourCountMap.size
      logger.info("日期[{}]需要抽取的数量为[{}]", day, dayExtractCount)

      // 遍历每小时的数据
      hourCountMap.foreach(hourCount => {
        val hour: String = hourCount._1
        val count: Long = hourCount._2
        // 每小时需要抽取的数量
        var hourExtractCount: Long = (count.toDouble / daySessionCounts * dayExtractCount).toInt
        logger.info("小时[{}]需要抽取的数量为[{}]", hour, hourExtractCount)
        if (hourExtractCount > count) {
          hourExtractCount = count
        }

        var extractIndexs: ListBuffer[Int] = hourExtractMap.getOrElse(hour, null)
        if (null == extractIndexs) {
          extractIndexs = ListBuffer()
          hourExtractMap += (hour -> extractIndexs)
        }
        for (_ <- 0 until hourExtractCount.toInt) {
          var extractIndex: Int = random.nextInt(count.toInt)
          while (extractIndexs.contains(extractIndex)) {
            extractIndex = random.nextInt(count.toInt)
          }
          extractIndexs.append(extractIndex)
        }
      })
    })
    println("dayHourExtarctMap:" + dayHourExtarctMap)
    dayHourExtarctMap
  }

  /**
    * 获取sparkSession
    * 如果是本地环境，就生成SqlContext
    * 如果是生产环境，就生成HiveContext
    *
    * @return
    */
  private def getSparkSession: SparkSession = {
    val sparkLocal: Boolean = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL)
    val sparkSession: SparkSession.Builder = SparkSession.builder.appName(Constants.SPARK_APP_NAME_SESSION).master("local[1]")
    if (sparkLocal) {
      sparkSession.getOrCreate()
    } else {
      sparkSession.enableHiveSupport().getOrCreate()
    }
  }
}
